package com.mjf.day7

import com.mjf.day3.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}

/**
 * 输出表计算的结果到文件
 *
 * 动态表和持续查询
 *   这是流处理程序转换成表结构的核心概念
 *
 * 更新模式
 *   Append仅追加（需要修改的数据不适用）
 *   Retract撤回（ADD和Delete）
 *   Upsert（Upsert和Delete）
 */
object OutputExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    // 读取数据到DataStream
    val inputStream: DataStream[String] = env.readTextFile("D:\\coding\\idea\\flink-stu\\src\\main\\input\\sensor.txt")

    val sensorStream: DataStream[SensorReading] = inputStream.map {
      data =>
        val words: Array[String] = data.split(",")
        val sensorId: String = words(0)
        val ts: Long = words(1).toLong
        val temp: Double = words(2).toDouble
        SensorReading(sensorId, ts, temp)
    }

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // 将DataStream转换为Table，同时设置Schema(基于名称，可以更换字段位置和重命名)
    val sensorTable: Table =
      tableEnv.fromDataStream(sensorStream, 'id, 'temperature as 'temp, 'timestamp as 'ts)

    // 基于位置（不能更换字段位置）
//    tableEnv.fromDataStream(sensorStream, 'myId, 'ts)

    // 定义一张输出表，这就是要写入数据的TableSink
    tableEnv
      .connect(new FileSystem().path("D:\\coding\\idea\\flink-stu\\src\\main\\output\\out.txt"))
      .withFormat(new Csv)
      .withSchema(
        new Schema()
          .field("id", DataTypes.STRING())
          .field("temp", DataTypes.DOUBLE())
//          .field("ts", DataTypes.BIGINT())
      )
      .createTemporaryTable("outputTable")

    // 将转换后的结果写出
    sensorTable
      .select("id, temp")
      .filter("id = 'sensor_1'")
      .insertInto("outputTable")

    // 将聚合后的结果写出
    // 报错，`CsvTableSink`继承了`AppendStreamTableSink`，不能写入聚合结果
//    sensorTable
//        .groupBy('id)
//        .select('id, 'id.count)
//        .insertInto("outputTable")


//    sensorTable.printSchema()
//    sensorTable.toAppendStream[(String,Double,Long)].print()


    env.execute("OutputExample")

  }
}

